package phoneData

import java.lang

import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.{Mapper, Reducer}
import org.apache.hadoop.util.ToolRunner

object MapReduce {

  class InputMapper extends Mapper[Object, Text, LongWritable, FlowBean] {

    private val word = new LongWritable()

    override def map(key: Object, value: Text, context: Mapper[Object, Text, LongWritable, FlowBean]#Context): Unit = {
      val line = value.toString.split("\t")
      val up = line(8).toLong
      val down = line(9).toLong
      word.set(line(1).toLong)
      context.write(word, FlowBean(up, down, 0))
    }
  }

  class OutputReducer extends Reducer[LongWritable, FlowBean, LongWritable, FlowBean] {
    override def reduce(key: LongWritable, values: lang.Iterable[FlowBean], context: Reducer[LongWritable, FlowBean, LongWritable, FlowBean]#Context): Unit = {
      import scala.jdk.CollectionConverters._
      val tuple = values.iterator()
        .asScala
        .foldLeft((0L, 0L))((nums, line) => {
          (nums._1 + line.upFlow, nums._2 + line.downFlow)
        })
      context.write(key, FlowBean(tuple._1, tuple._2, tuple._1 + tuple._2))
    }
  }


  @throws[Exception]
  def main(args: Array[String]): Unit = {
    // System.setProperty("hadoop.home.dir", "D:\\development\\hadoop-3.0.0")
    System.exit(ToolRunner.run(new MRDriver, args))
  }

}
